//package com.tjhk.demo.kafka.config;
//
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Qualifier;
//import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
//import org.springframework.boot.context.properties.ConfigurationProperties;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.context.annotation.Primary;
//import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
//import org.springframework.kafka.config.KafkaListenerContainerFactory;
//import org.springframework.kafka.core.ConsumerFactory;
//import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
//import org.springframework.kafka.core.DefaultKafkaProducerFactory;
//import org.springframework.kafka.core.KafkaTemplate;
//import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
//import org.springframework.kafka.listener.ContainerProperties;
//
///**
// * 第一个kafka配置
// *
// */
//@Configuration
//public class FirstKafkaConfig {
//
//    /**
//     * 读取第一个kafka配置
//     * Primary注解表示默认以这个为准
//     *
//     * @return 第一个kafka配置
//     */
//    @Primary
//    @ConfigurationProperties(prefix = "spring.kafka.first")
//    @Bean
//    public KafkaProperties firstKafkaProperties() {
//        return new KafkaProperties();
//    }
//
//    /**
//     * 构建第一个kafka的生产者发送template
//     *
//     * @param firstKafkaProperties 第一个kafka配置
//     * @return 第一个kafka的生产者发送template
//     */
//    @Primary
//    @Bean
//    public KafkaTemplate<String, byte[]> firstKafkaTemplate(
//            @Autowired @Qualifier("firstKafkaProperties") KafkaProperties firstKafkaProperties) {
//        return new KafkaTemplate<>(firstProducerFactory(firstKafkaProperties));
//    }
//
//    /**
//     * 构建第一个kafka的消费者监听容器工厂
//     *
//     * @param firstKafkaProperties 第一个kafka配置
//     * @return 第一个kafka的消费者监听容器工厂
//     */
//    @Bean("firstKafkaListenerContainerFactory")
//    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, byte[]>>
//    firstKafkaListenerContainerFactory(@Autowired @Qualifier("firstKafkaProperties") KafkaProperties firstKafkaProperties) {
//        ConcurrentKafkaListenerContainerFactory<String, byte[]> factory =
//                new ConcurrentKafkaListenerContainerFactory<>();
//        factory.setConsumerFactory(firstConsumerFactory(firstKafkaProperties));
//        // 开启批量消费
//        factory.setBatchListener(true);
//        // 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
//        //RECORD,
//        //// 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
//        //BATCH,
//        //// 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
//        //TIME,
//        //// 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
//        //COUNT,
//        //// TIME |　COUNT　有一个条件满足时提交
//        //COUNT_TIME,
//        //// 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
//        //MANUAL,
//        //// 手动调用Acknowledgment.acknowledge()后立即提交
//        //MANUAL_IMMEDIATE,
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
//        return factory;
//    }
//
//    /**
//     * 新建第一个kafka的消费者工厂
//     *
//     * @param firstKafkaProperties 第一个kafka配置
//     * @return 第一个kafka的消费者工厂
//     */
//    private ConsumerFactory<? super String, ? super byte[]> firstConsumerFactory(KafkaProperties firstKafkaProperties) {
//        return new DefaultKafkaConsumerFactory<>(firstKafkaProperties.buildConsumerProperties());
//    }
//
//    /**
//     * 新建第一个kafka的生产者工厂
//     *
//     * @param firstKafkaProperties 第一个kafka配置
//     * @return 第一个kafka的生产者工厂
//     */
//    private DefaultKafkaProducerFactory<String, byte[]> firstProducerFactory(KafkaProperties firstKafkaProperties) {
//        return new DefaultKafkaProducerFactory<>(firstKafkaProperties.buildProducerProperties());
//    }
//}
